package com.ruiyian.netty.nettyUdpserver;

import com.sun.javafx.collections.MappingChange;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollDatagramChannel;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * netty实现udpserver服务
 */
@Component
@Data
@NoArgsConstructor
@AllArgsConstructor
public class NettyUdpServer {
    private static final Logger LOG = LoggerFactory.getLogger(NettyUdpServer.class);

    private EventLoopGroup bossLoopGroup;

    private Channel serverChannel;

    /**
     * netty初始化
     */
    public void init(int port) {

        LOG.info("Epoll.isAvailable():{}", Epoll.isAvailable());


        //表示服务器连接监听线程组，专门接受 accept 新的客户端client 连接
        bossLoopGroup = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();

        try {
            //1、创建netty bootstrap 启动类
            Bootstrap serverBootstrap = new Bootstrap();
            //2、设置boostrap 的eventLoopGroup线程组
            serverBootstrap.group(bossLoopGroup)
                    //3、设置NIO UDP连接通道
                    .channel(Epoll.isAvailable() ? EpollDatagramChannel.class : NioDatagramChannel.class)
                    //4、设置通道参数 SO_BROADCAST广播形式
                    .option(ChannelOption.SO_BROADCAST, true)
                    .option(ChannelOption.SO_RCVBUF, 1024 * 1024)
                    //5、设置处理类 装配流水线
                    .handler(new NettyUdpHandler());

            // linux平台下支持SO_REUSEPORT特性以提高性能
            if (Epoll.isAvailable()) {
                LOG.info("SO_REUSEPORT");
                serverBootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
            }

            // 如果支持epoll则说明是Linux版本，则利用SO_REUSEPORT创建多个线程
            if (Epoll.isAvailable()) {
                // linux系统下使用SO_REUSEPORT特性，使得多个线程绑定同一个端口
                int cpuNum = Runtime.getRuntime().availableProcessors();
                LOG.info("using epoll reuseport and cpu:" + cpuNum);
                for (int i = 0; i < cpuNum; i++) {
                    LOG.info("worker-{} bind", i);
                    //6、绑定server，通过调用sync（）方法异步阻塞，直到绑定成功
                    ChannelFuture future = serverBootstrap.bind(port).sync();
                    if (!future.isSuccess()) {
                        LOG.error("bootstrap bind fail port is " + port);
                        throw new Exception(String.format("Fail to bind on [host = %s , port = %d].", "192.168.2.128", port), future.cause());
                    } else {
                        LOG.info("bootstrap bind success ");
                    }
                }
            } else {
                ChannelFuture future = serverBootstrap.bind(port).sync();
                if (!future.isSuccess()) {
                    LOG.error("bootstrap bind fail port is " + port);
                    throw new Exception(String.format("Fail to bind on [host = %s , port = %d].", "127.0.0.1", port), future.cause());
                } else {
                    LOG.info("bootstrap bind success ");
                }
            }

        } catch (Exception e) {
            LOG.error("报错了，错误原因:{}", e.getMessage(), e);
        }
    }
}
